package com.cdel.akka.route;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.Terminated;
import akka.actor.UntypedActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.routing.ActorRefRoutee;
import akka.routing.RoundRobinRoutingLogic;
import akka.routing.Routee;
import akka.routing.Router;

//路由策略：包括轮询actor进行消息发送（RoundRobinRoutingLogic），广播策略（broadcastRoutingLogic），随机投递策略（RandomRoutingLogic）
//空闲actor优先投递策略（SmallestMailboxRoutingLogic）
public class WatchActor extends UntypedActor {

	private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
	public Router router;
	private AtomicInteger i = new AtomicInteger(5);
	{
		List<Routee> routees = new ArrayList<>();
		for(int i=0; i<5; i++){
			ActorRef worker = getContext().actorOf(Props.create(MyWorker.class), "worker_" + i);
			getContext().watch(worker);
			routees.add(new ActorRefRoutee(worker));
		}
		router = new Router(new RoundRobinRoutingLogic(), routees);
	}

	@Override
	public void onReceive(Object msg) throws Throwable {
		//当有消息需要传递给这5个MyWorker时，只需要将消息投递给这个Router即可。Router会根据给定的消息路由策略进行消息投递。当一个MyWorker停止工作后，
		//还可以简单的将其从工作组中移出，在这里，如果发现系统中没有可用的actor，就会直接关闭系统。
		if(msg instanceof MyWorker.Msg){
			router.route(msg, getSender());
		}else if(msg instanceof Terminated){
			router = router.removeRoutee(((Terminated)msg).actor());
			System.out.println(((Terminated)msg).actor().path() + " is closed, routees=" + i.decrementAndGet());
			if(i.get() == 0){
				System.out.println("Close system");
				RouteMain.flag.send(false);
				getContext().system().shutdown();
			}
		}else{
			unhandled(msg);
		}
	}

}
